package com.project.base.kafka.producer;

import com.project.base.common.Constants;
import com.project.base.core.kafka.IKafkaSender;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import java.io.Serializable;
import java.util.Properties;

public class KafkaProducer implements IKafkaSender, InitializingBean, DisposableBean, Serializable {
    private static final long serialVersionUID = 1195108028882834731L;
    private static final Logger LOGGER = Logger.getLogger(KafkaProducer.class);
    public transient static Producer<String, String> producer;

    @Override
    public void send(String[] topics, String msg) {
        if (topics == null || topics.length == 0) {
            LOGGER.info("Topic is null, send failed");
            return;
        }
        if (producer == null) {
            synchronized (KafkaProducer.class) {
                if (producer == null) {
                    try {
                        afterPropertiesSet();
                    } catch (Exception e) {
                        LOGGER.error(e.getMessage(), e);
                        return;
                    }
                }
            }
        }

        for (int i = 0; i < topics.length; i++) {
            KeyedMessage<String, String> km = new KeyedMessage<String, String>(topics[i], msg);
            producer.send(km);
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Properties properties = new Properties();
        properties.put("metadata.broker.list", Constants.getInstance().getValue("client.service.kafka.metadata.broker.list"));
        properties.put("producer.type", Constants.getInstance().getValue("client.service.kafka.producer.type"));
        properties.put("compression.codec", Constants.getInstance().getValue("client.service.kafka.compression.codec"));
        properties.put("serializer.class", Constants.getInstance().getValue("client.service.kafka.serializer.class"));
        ProducerConfig config = new ProducerConfig(properties);
        producer = new Producer<String, String>(config);
    }

    @Override
    public void destroy() throws Exception {
        if (producer != null) {
            producer.close();
        }
    }
}
